#pragma once

/*
 * The anet tcp connection.
 */
#include <any>
#include <initializer_list>
#include <memory>
#include <queue>
#include <atomic>
#include "event_loop.hpp"
#include "define.hpp"
#include "anet.hpp"
#include "log.h"
#include "asio/detail/noncopyable.hpp"
#include "pool/objpool.h"
#include "util/allocator.hpp"

namespace anet {
	namespace tcp {
		class CConnection;
		typedef std::pair<bool, std::string> sendDataPair;
		typedef std::enable_shared_from_this<CConnection> shareFromThisConnection;
		typedef std::function<void(connSharePtr)> removeFuncCallback;

		class CConnection : public shareFromThisConnection, asio::noncopyable {
		public:
			CConnection(asioIOContext &ioContext, CEventLoop &loop) :
				m_ioContext(ioContext), m_loop(loop),
				m_sock(ioContext), m_readSize(0), m_codec(nullptr), 
				m_session(nullptr), m_closed(true), m_peerPort(0), 
				m_peerIP(""), m_objId(0), m_removeCallback(nullptr) {

				// initialize the receiving buffer.
				m_readBuff = new char[gMaxReadBuffLength];

				// reserve the sending buffer.
				m_sendBuff.second.reserve(gMaxWriteBuffLength);
				m_willSendBuff.second.reserve(gMaxWriteBuffLength);
			}

			virtual ~CConnection() {
				if (!m_closed) {
					this->closeSocket();
					m_closed = true;
				}

				assert(m_readBuff != nullptr && "receiving buffer is null");
				if (m_readBuff == nullptr) {
					return;
				}
				delete []m_readBuff;
				m_readBuff = nullptr;
			}

		public:
			void setAny(std::any any) {
				this->m_attach = any;
			}
			std::any getAny() {
				return m_attach;
			}

			void SetObjId(unsigned int objId) {
				m_objId = objId;
			}
			unsigned int GetObjId() const {
				return m_objId;
			}

			void setRemoveFunc(removeFuncCallback func) {
				m_removeCallback = std::move(func);
			}

			// setLinger sets socket linger time.
			// 0 is close the linger mode, else open linger mode.
			bool setLinger(int s) {
				asio::error_code error_code;
				if (s > 0) {
					m_sock.set_option(asio::socket_base::linger(true, s), error_code);
				} else {
					m_sock.set_option(asio::socket_base::linger(false, 0), error_code);
				}
				return !error_code;
			}

			// send sends share_ptr<string> message.
			void send(std::shared_ptr<std::string> msg) {
				if (m_closed) { return; }

				// Empty string is for connection close.
				if (msg->empty()) { return; }
				
				// share pointer.
				auto self = shared_from_this();

				// post to m_ioContext message.
				asio::post(m_ioContext, [this, self, msg]() {
					this->initSendBuffer(std::move(*msg));
				});
			}
			inline void send(const std::string& data) {
				this->send(data.c_str(), data.length());
			}
			// send sends binary data and its length.
			inline void send(const char *data, size_t len) {
				this->Send(data, len);
			}
			inline void send(std::string&& msg) {
				this->send(msg.c_str(), msg.length());
			}

			// send with memory pool.
			inline void Send(const char* msg, size_t len) {
				try {
					if (!msg) {
						return;
					}
					if (m_closed) { return; }

					// std::string message(msg, len);
					anet::utils::PoolString message(msg, len);

					// share pointer.
					auto self = shared_from_this();

					// post to m_ioContext message.
					asio::post(m_ioContext, [this, self, strMsg = std::move(message)]() {
						this->initSendBuf(strMsg.c_str(), strMsg.length());
					});
				} catch (const std::bad_alloc& e) {
					LogCrit("Memory allocation failed: %s", e.what());
				}
			}

			// close closes the socket which will try to send out all the messages in the queue.
			void close() {
				if (m_closed) { return; }

				// Try to send empty data to denote close the connection.
				auto self = shared_from_this();
				asio::post(m_ioContext, [this,self,msg=std::move(std::string{ "" })]() {
					this->initSendBuffer(msg);
				});
			}

		public:
			// getRemoteAddr gets remote ip
			const char* getRemoteAddr() const {
				return m_peerIP.c_str();
			}

			// getRemotePort gets remote port
			unsigned short getRemotePort() const {
				return m_peerPort;
			}

			// getSocket gets socket.
			asioSocket& getSocket() {
				return m_sock;
			}

			// doStart starts connection established.
			void doStart(ISession *session, ICodec *codec) {
				assert(session != nullptr && codec != nullptr);
				if (session == nullptr || codec == nullptr) {
					LogCrit("session or codec is nullptr");
					return;
				}

				{// Set keep alive.
					asio::error_code ec;
					m_sock.set_option(asio::socket_base::keep_alive(true), ec);
					if (ec) {
						LogWarn("set socket keep alive error:%d", ec.value());
					}
				}

				{// Set no delay = true, just do NOT use Nagle.
					asio::error_code ec;
					m_sock.set_option(asio::ip::tcp::no_delay(true), ec);
					if (ec) {
						LogWarn("set socket no delay error:%d", ec.value());
					}
				}

				// set time out
				this->setTimeout();

				// set open flag.
				m_closed = false;
				m_session = session;
				m_codec = codec;

				// save ip and port.
				m_peerIP = m_sock.remote_endpoint().address().to_string();
				m_peerPort = m_sock.remote_endpoint().port();

				// connected callback.
				m_session->onConnected(shared_from_this());

				// try to set reading message hook.
				this->postRead();
			}

			// closeSocket closes socket.
			void closeSocket() {
				m_sock.cancel();
				asio::error_code ec;
				m_sock.shutdown(asioSocket::shutdown_both, ec);

				asio::error_code ignore_err;
				m_sock.close(ignore_err);
			}

		protected:
			// setTimeout sets time out flag.
			void setTimeout() {
				// the timeout value
				unsigned int timeout_milli = 10000;
				auto sock = m_sock.native_handle();

				// platform-specific switch
            #if defined _WIN32 || defined WIN32 || defined OS_WIN64 || defined _WIN64 || defined WIN64 || defined WINNT
				int timeout = timeout_milli;
				setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout, sizeof(timeout));
				setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (const char*)&timeout, sizeof(timeout));
            #else
				struct timeval tv;
				tv.tv_sec = timeout_milli / 1000;
				tv.tv_usec = (timeout_milli % 1000) * 1000;
				setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
				setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
            #endif
			}

			// doClose posts to close all resource.
			void doClose() {
				bool closed = false;
				if (!m_closed.compare_exchange_strong(closed, true)) {
					return;
				}

				auto self = shared_from_this();
				asio::post(m_ioContext, [this, self]() {
					m_session->onTerminate();
					this->closeSocket();
					m_session->release();
					m_session = nullptr;

					// call removing callback function(!!must be the last line!!)
					if (m_removeCallback != nullptr) {
						m_removeCallback(self);
					}
				});
			}

			// isDisplayNetErrorCode is whether display the error information.
			// display error
			// all the following errors are just ignore.
			// do not display asio::error::operation_aborted
			// asio::error:eof,
			// asio::error::connection_reset
			bool isDisplayNetErrorCode(const asio::error_code ec) const {
				if (ec != asio::error::operation_aborted &&
					ec != asio::error::eof &&
					ec != asio::error::connection_reset &&
					ec != asio::error::bad_descriptor &&
					ec != asio::error::connection_aborted &&
					ec != asio::error::timed_out) {
					return true;
				} else {
					return false;
				}
			}

			// postSend posts sending request.
			void postSend() {
				if (m_closed) {
					return;
				}

				assert(m_sendBuff.first && "send flag is true");
				if (m_sendBuff.second.empty()) {
					this->doClose();
					return;
				}

				// post socket buffer to write message.
				auto self = shared_from_this();
				asio::async_write(m_sock, 
					asio::buffer(m_sendBuff.second.data(), m_sendBuff.second.length()), 
					[this,self](asio::error_code ec, std::size_t sendBytes) {
					if (!ec) {
						if (m_willSendBuff.first) {
							// if it is asked to close the connection, then set the sending content to be empty.
							m_sendBuff.first = true;
							m_sendBuff.second.clear();
							this->postSend();
						} else if (!m_willSendBuff.second.empty()) {
							// try to send the m_willSendBuff message.
							m_sendBuff.first = true;
							m_sendBuff.second.swap(m_willSendBuff.second);
							m_willSendBuff.second.clear();
							this->postSend();
						} else {
							// if there is no message, just wait for the next message.
							m_sendBuff.first = false;
							m_sendBuff.second.clear();
						}
					} else {
						// just close if error occurs.
						if (isDisplayNetErrorCode(ec)) {
							LogWarn("send to %s:%d error:%d(%s)", 
								m_peerIP.c_str(), 
								m_peerPort, ec.value(), 
								ec.message().c_str()
							);
						}
						this->doClose();
					}
				});
			}

			// postRead posts reading requirement.
			void postRead() {
				if (m_closed) { return; }

				auto self = shared_from_this();
				char *recvBuff = m_readBuff + m_readSize;
				int recvSize = gMaxReadBuffLength - m_readSize;
				m_sock.async_read_some(asio::buffer(recvBuff, recvSize),
					[this,self](asio::error_code ec, std::size_t len) {
					if (!ec) {
						// return if the connection was closed.
						if (m_session == nullptr || !m_sock.is_open()) {
							return;
						}

						m_readSize += int(len);
						// try to parse all the packets.
						int allParseLen = this->tryProcessAllPackets();
						if (allParseLen < 0) {
							LogCrit("receive an invalid packet from %s:%d", m_peerIP.c_str(), m_peerPort);
							this->doClose();
							return;
						} else if (allParseLen == 0) {
							// there is no read buffer left,just close it
							if (m_readSize >= gMaxReadBuffLength) {
								LogCrit("there is no readable buffer left.");
								this->doClose();
								return;
							}
						} else {
							// correct the receiving buffer's reading position.
							if (allParseLen == m_readSize) {
								m_readSize = 0;
							} else if (allParseLen < m_readSize) {
								size_t allReadSize = size_t(m_readSize);
								size_t parseSize = size_t(allParseLen);
								size_t leftSize = allReadSize - parseSize;
								memmove(m_readBuff, m_readBuff + allParseLen, leftSize);
								m_readSize -= allParseLen;
							} else {
								assert(false && "parse return invalid value");
							}
						}
						this->postRead();
					} else {
						if (isDisplayNetErrorCode(ec)) {
							LogInfo("read %s:%d message error:%d(%s)", 
								m_peerIP.c_str(), 
								m_peerPort, 
								ec.value(), 
								ec.message()
							);
						}
						this->doClose();
					}
				});
			}

			// tryProcessAllPackets tries to parse all the packets and returns all the packets len.
			inline int tryProcessAllPackets() {
				int parseAllPacketsLen = 0;
				char *currentBuff = m_readBuff;
				int leftSize = m_readSize;

				// max parse count.
				const int gMaxParseCount = 1024;
				int count = gMaxParseCount;
				for (; count >= 0; count--) {
					int packetLen = m_codec->parsePacket(currentBuff, leftSize);
					if (packetLen == codecRet::retError) {
						// return invalid packet.
						return -1;
					} else if (packetLen > 0) {
						assert(packetLen <= leftSize && "parse data size error");
						m_session->onRecv(currentBuff, packetLen);

						parseAllPacketsLen += packetLen;

						// next data buffer and its size.
						currentBuff = currentBuff + packetLen;
						leftSize -= packetLen;
						if (leftSize <= 0) {
							return parseAllPacketsLen;
						}
					} else {
						// return all packets len
						return parseAllPacketsLen;
					}
				}
				return parseAllPacketsLen;
			}

			// initSendBuffer initializes sending buffer.
			void initSendBuffer(const std::string &msg) {
				this->initSendBuf(msg.c_str(), msg.length());
			}
			void initSendBuf(const char *pBuf, size_t len) {
				if (0 == len) {
					// try to close receive mode if it closes the socket.
					asio::error_code ec;
					m_sock.shutdown(asioSocket::shutdown_receive, ec);
				}

				// if it has been set close flag. so just return
				if (m_willSendBuff.first) {
					return;
				}

				bool post = false;
				if (!m_sendBuff.first) {
					// if there is no message to send, append to sendBuffer, and post to send.
					m_sendBuff.second.append(pBuf,len);
					m_sendBuff.first = true;
					post = true;
				} else {
					// if it is sending, so add to willSendBuff to wait for sending.
					if (0 == len) {
						m_willSendBuff.first = true;
					} else {
						m_willSendBuff.first = false;
						m_willSendBuff.second.append(pBuf,len);
					}
					post = false;
				}
				if (post) {
					this->postSend();
				}
			}

		private:
			// asio ioContext;
			asioIOContext    &m_ioContext;
			CEventLoop       &m_loop;

			// tcp socket.
			asioSocket       m_sock;

			// receiving buffer and its current size.
			char*            m_readBuff;
			int              m_readSize;

			// user callback interface, must be initialized outside.
			ICodec           *m_codec{ nullptr };
			ISession         *m_session{ nullptr };

			// close flag(closed:true,else false)
			std::atomic_bool m_closed{ true };

			// remote peer address;
			unsigned short   m_peerPort{ 0 };
			std::string      m_peerIP{ "" };

			// current send buffer{sending-flag, sending-buffer}
			sendDataPair     m_sendBuff{ false,"" };

			// next send buffer{close-flag, next-buffer}
			sendDataPair     m_willSendBuff{ false,"" };

			// object id.
			unsigned int     m_objId{ 0 };

			// remove callback(close callback).
			removeFuncCallback m_removeCallback{ nullptr };

			// any attach.
			std::any          m_attach;
		};
	}
}
